package com.atguigu.api3

import com.atguigu.api.SensorReading
import org.apache.flink.streaming.api._
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
 *
 * @description: 在processTime处理模式下,对延时数据可以直接将时间延时若干单位时间,
 *               在eventTime处理模式下,对延时数据就可以通过watermark机制来处理,watermark用于表示timestamp小于Watermark的数据都已经达到.用来让程序自己平衡延迟和结果正确性
 *               watermark是一条特殊的数据记录
 *               必须单调递增,以确保任务的事件时间时钟都在向前推进,而不是在后退,watermark与数据的时间戳相关,
 *               首先确定延时3s[数据中最大差值],0-5s的窗口 6-10 首先数据都各自放到相应
 *               的桶里面,然后来了8 8-3=5那么0-5这个窗口就可以关闭计算了,后来的4则丢弃,有了watermark后延时处理数据关闭窗口也是以watermark为基准的.如210-225 225-245 来了一个285s的数据那么当前watermark为285-3=282,
 *               此时282距离225s为57s还达不到关窗的条件,如果来到了289s数据进来了.那么watermark=286 286-225=61延时也满足了直接关窗聚合计算了.
 * @time: 2021-03-15 15:35
 * @author: baojinlong 
 **/
object WatermarkDemoTestFromSocket {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    //设置事件时间机制,watermark仅仅在事件时间语义下处理策略
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 设置生成watermark周期时间
    environment.getConfig.setAutoWatermarkInterval(500)
    // 从文本读取
    //val inputStreamFromFile: DataStream[String] = environment.readTextFile("E:/big-data/FlinkTutorial/src/main/resources/sensor.data")
    val inputStreamFromFile: DataStream[String] = environment.socketTextStream("localhost", 1000)
    // 基本转换操作
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      })
      // 指定事件时间是从事件数据中获取同时是递增的,这是递增数据.watermark就不需要定义
      //.assignAscendingTimestamps(_.timestamp * 1000L)
      // 针对乱序数据可以选择周期性和间断性watermark,一般选择前者考虑到性能 延时30ms,如果还有延时数据则还可以通过设置延时1min,剩下的乱序还可以作为侧输出流单独处理.code48 49
      // flink处理延时数据有三种机制,首先设置很小的延时得到近似数据,设置允许处理延时1min数据,最后剩下数据可以放入到侧输出流中
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.milliseconds(30)) {
        override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000L
      })

    // 定义侧输出流
    val laterOutputData = new OutputTag[(String, Double, Long)]("later1MinData")
    // 每15s统计一次,窗口内各传感器所有温度的最小值,以及最新的时间戳
    val resultStream: DataStream[(String, Double, Long)] = dataStream
      .map(data => (data.id, data.temperature, data.timestamp))
      .keyBy(_._1)
      .timeWindow(Time.seconds(15)) // 滚动窗口
      .allowedLateness(Time.minutes(1)) // 整体延时等待1min
      // 剩下所有数据放入到侧输出流中
      .sideOutputLateData(laterOutputData)
      .reduce((curRes, newData) => (curRes._1, curRes._2.min(newData._2), newData._3))
    // 获取侧输出流
    resultStream.getSideOutput(laterOutputData).print("laterOutputData")
    resultStream.print("window event time")
    environment.execute("xx")
  }

}
